package com.atguigu.app.dwd.db;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * ClassName: BaseLogApp
 * Package: com.atguigu.app.dwd
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/15 19:03
 * @Version 1.0
 */
//数据流：web/app -> 日志服务器(log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序：Mock -> 文件  -> Flume -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //1.环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS),Time.of(3L,TimeUnit.MINUTES)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //读取业务主流
        String topic = "topic_log";
        String groupId = "dwd_app";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        kafkaDS.print("kafkaDS>>>>");
        // 3.过滤数据并转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        out.collect(jsonObject);
                    } catch (JSONException e) {
                        System.out.println("脏数据：" + value);
                    }
                }
            }
        });

        //TODO 4.按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //TODO 5.使用状态编程对新老用户字段做校验
        SingleOutputStreamOperator<JSONObject> newFlagJsonDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastVisitState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastVisitState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last_visit", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                //获取is_new标记、状态数据、当前数据中的日期 更新状态
                String isNew = value.getJSONObject("common").getString("is_new");
                String lastValue = lastVisitState.value();
                Long ts = value.getLong("ts");
                String curDt = DateFormatUtil.toDate(ts);

                if ("1".equals(isNew)) {
                    if (lastValue == null || lastValue.equals(curDt)){
                        lastVisitState.update(curDt);
                    } else value.getJSONObject("common").put("is_new", "0");
                } else {
                    if (lastValue == null)
                        lastVisitState.update("1970-01-01");
                }

                return value;
            }
        });
        //TODO 6.使用侧输出流进行分流处理  主流为 play 4个侧输出流   start  display  action error page
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        OutputTag<String> actionTag = new OutputTag<String>("action"){};
        OutputTag<String> displayTag = new OutputTag<String>("display"){};
        OutputTag<String> errorTag = new OutputTag<String>("error"){};
        OutputTag<String> pageTag = new OutputTag<String>("page"){};

        SingleOutputStreamOperator<String> playDS = newFlagJsonDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                //尝试获取错误信息
                String err = value.getString("err");
                if (err != null) {
                    ctx.output(errorTag, value.toString());
                    value.remove("err");
                }
                //尝试获取启动和页面信息
                String start = value.getString("start");
                String page = value.getString("page");
                if (start != null) {
                    ctx.output(startTag, value.toString());
                }
                //尝试获取页面信息
                else if(page !=null)
                {
                    Long ts = value.getLong("ts");
                    String pageId = value.getJSONObject("page").getString("page_id");
                    //尝试获取曝光日志
                    JSONArray displays = value.getJSONArray("display");
                    if (displays != null && displays.size() > 0) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("page_id", pageId);
                            display.put("ts", ts);
                            ctx.output(displayTag, value.toString());
                        }
                    }
                    //尝试获取动作数据
                    JSONArray actions = value.getJSONArray("action");
                    if (actions != null && actions.size() > 0) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("page_id", pageId);
                            ctx.output(actionTag, value.toString());
                        }
                    }
                    value.remove("display");
                    value.remove("action");
                    ctx.output(pageTag, value.toString());
                }
                else //其余为播放日志
                     {
                    out.collect(value.toString());


                }
            }
        });
        //获取其他日志数据
        DataStream<String> startDS = playDS.getSideOutput(startTag);
        DataStream<String> displayDS = playDS.getSideOutput(displayTag);
        DataStream<String> actionDS = playDS.getSideOutput(actionTag);
        DataStream<String> errorDS = playDS.getSideOutput(errorTag);
        DataStream<String> pageDS = playDS.getSideOutput(pageTag);

        //TODO 7.将数据写出
        pageDS.print("pageDS>>>>");
        startDS.print("startDS>>>>");
        displayDS.print("displayDS>>>>");
        actionDS.print("actionDS>>>>");
        errorDS.print("errorDS>>>>");
        playDS.print("playDS>>>>");

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String play_topic = "dwd_traffic_play_log";

        pageDS.addSink(KafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(KafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(KafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(KafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(KafkaUtil.getFlinkKafkaProducer(error_topic));
        playDS.addSink(KafkaUtil.getFlinkKafkaProducer(play_topic));

        env.execute();



    }
}
